package com.xian.java.sink;

import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.*;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class MyFileBatchSink {
    public static void main(String[] args) throws Exception {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataSet<String> ds = env.readTextFile("J:\\test\\wc.txt");

        FlatMapOperator<String, String> Operator1 = ds.flatMap(new RichFlatMapFunction<String, String>() {
            @Override
            public void flatMap(String line, Collector<String> out) throws Exception {
                String[] arr = line.split(" ");
                for (String word : arr) {
                    out.collect(word);
                }
            }
        });


        MapOperator<String, Tuple2<String, Integer>> operator2 = Operator1.map(new RichMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String word) throws Exception {
                return new Tuple2<String, Integer>(word, 1);
            }
        });

        UnsortedGrouping<Tuple2<String, Integer>> groupBy = operator2.groupBy(0);
        AggregateOperator<Tuple2<String, Integer>> aggregateOperator = groupBy.sum(1);



        aggregateOperator.print();

        aggregateOperator.writeAsText("J:\\test\\out2", FileSystem.WriteMode.OVERWRITE);
        aggregateOperator.writeAsCsv("J:\\test\\out3", FileSystem.WriteMode.OVERWRITE);


        env.execute();
    }
}
